package com.chinamobile.iot.service.impl;

import com.alibaba.fastjson.JSON;
import com.chinamobile.iot.config.KafkaTopicConfiguration;
import com.chinamobile.iot.kafka.KafkaCommandProducer;
import com.chinamobile.iot.mqtt.spi.impl.ProtocolProcessor;
import com.chinamobile.iot.mqtt.spi.processor.CommandState;
import com.chinamobile.iot.mqtt.spi.processor.DataProcessor;
import com.chinamobile.iot.mqtt.spi.security.IAuthorizator;
import com.chinamobile.iot.service.BrokerService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.Base64Utils;

import java.nio.ByteBuffer;


@Service
public class BrokerServiceImpl implements BrokerService {

    private Logger logger = LoggerFactory.getLogger(getClass());
    @Autowired
    private ProtocolProcessor processor;
    /**
     * kafkatopic的配置
     */
    @Autowired
    private KafkaTopicConfiguration kafkaTopic;
    @Autowired
    private KafkaCommandProducer kafkaCommandProducer;
    @Autowired
    private IAuthorizator authorizator;
    @Autowired
    private DataProcessor dataProcessor;

    @Override
    public boolean isOnline(String mqttId) {
        return processor.isOnline(mqttId);
    }

    @Override
    public int getClientCount() {
        return processor.getClientCount();
    }

    @Override
    public boolean sendCommand(String mqttId, String commandId, String type, String command) {
        logger.info("begin send command to device:{}, commandId:{}, type:{}, command:{}", mqttId, commandId, type, command);
        // 接收命令的ClientId
        String clientId = processor.getClientId(mqttId);
        // 接收命令的DeviceId
        String realMqttId = mqttId;
        if (clientId == null) {
            realMqttId = authorizator.getProxy(mqttId, "write");
            if (realMqttId == null) {
                logger.info("the device {} is not connect to this broker or proxy on this broker, ", mqttId);
                return false;
            } else {
                clientId = processor.getClientId(realMqttId);
                if (clientId == null) {
                    logger.info("the device {} is not connect to this broker or proxy on this broker, ", mqttId);
                    return false;
                }
            }
        }
        logger.info("sendCommand() is called. clientId:{}, realMqttId:{}", clientId, realMqttId);

        byte[] payload;

        switch (type) {
            case "json": {
                payload = command.getBytes();
            }
            break;
            case "int": {
                Integer value = Integer.valueOf(command);
                payload = new byte[4];
                ByteBuffer byteBuffer = ByteBuffer.wrap(payload);
                byteBuffer.asIntBuffer().put(value);
            }
            break;
            case "double": {
                payload = new byte[8];
                Double value = Double.valueOf(command);
                ByteBuffer byteBuffer = ByteBuffer.wrap(payload);
                byteBuffer.asDoubleBuffer().put(value);
            }
            break;
            case "bin": {
                payload = Base64Utils.decodeFromString(command);
            }
            break;
            case "string": {
                payload = command.getBytes();
            }
            break;
            default: {
                logger.info("the command data format is invalid");
                return false;
            }
        }

        // 构造topic
        String topic = authorizator.getCmdTopic(clientId, mqttId);

        // 向客户端发送命令
        Integer msgId = processor.sendCommandQos1(clientId, realMqttId, commandId, topic, payload);
        if (msgId != null) {
            Object cmd = dataProcessor.createCommandEvent(clientId, mqttId, commandId, CommandState.SENDED);
            if (cmd != null) {
                kafkaCommandProducer.sendCommand(mqttId, kafkaTopic.cmd(), cmd);
                logger.info("send cmd sended to kafka :  {}", JSON.toJSONString(cmd));
            }
            return true;
        }
        return false;
    }

    @Override
    public boolean offlineDevice(String mqttId) {
        return processor.offlineMqttId(mqttId);
    }
}
